package com.bw.app.dws;


import com.bw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


public class Gd12_gender {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 1. 创建源表
        tEnv.executeSql("CREATE TABLE dwd_ua( " +
                "log_id STRING, " +
                "user_id STRING, " +
                "item_id STRING, " +
                "item_name STRING, " +
                "behavior_type STRING, " +
                "behavior_time TIMESTAMP(3), " +
                "search_keyword STRING, " +
                "session_id STRING, " +
                "stay_duration STRING, " +
                "page_url STRING, " +
                "category_id STRING, " +
                "category_path STRING, " +
                "keywords STRING, " +
                "description STRING, " +
                "price DOUBLE, " +
                "is_maternal STRING, " +
                "create_time TIMESTAMP(3), " +
                "WATERMARK FOR behavior_time AS behavior_time - INTERVAL '5' SECOND " +
                ")"  + MyKafkaUtil.getKafkaDDL("dwd_ua", "ua"));

// 2. 执行查询
        Table result = tEnv.sqlQuery(
                "WITH a1 AS (" +
                        "    SELECT " +
                        "        user_id, " +
                        "        item_id, " +
                        "        TUMBLE_START(behavior_time, INTERVAL '10' MINUTE) AS window_start, " +
                        "        behavior_time, " +
                        "        CASE behavior_type " +
                        "            WHEN 'purchase' THEN 1.0 " +
                        "            WHEN 'search'   THEN 0.8 " +
                        "            WHEN 'add_cart' THEN 0.7 " +
                        "            WHEN 'favorite' THEN 0.6 " +
                        "            ELSE 0.3 " +
                        "        END AS behavior_weight, " +
                        "        CASE " +
                        "            WHEN TIMESTAMPDIFF(SECOND, behavior_time, CAST(CURRENT_TIMESTAMP AS TIMESTAMP)) / (60*60*24) <= 30 THEN 1.0 " +
                        "            WHEN TIMESTAMPDIFF(SECOND, behavior_time, CAST(CURRENT_TIMESTAMP AS TIMESTAMP)) / (60*60*24) <= 60 THEN 0.5 " +
                        "            ELSE 0.2 " +
                        "        END AS time_decay " +
                        "    FROM dwd_ua " +
                        "    GROUP BY user_id, item_id, TUMBLE(behavior_time, INTERVAL '10' MINUTE), behavior_time, behavior_type " +
                        "), " +
                        "a2 AS ( " +
                        "    SELECT " +
                        "        item_id, " +
                        "        TUMBLE_START(behavior_time, INTERVAL '10' MINUTE) AS window_start, " +
                        "        CASE " +
                        "            WHEN keywords LIKE '%女%' AND keywords NOT LIKE '%男%' THEN 1.0 " +
                        "            WHEN keywords LIKE '%男%' AND keywords NOT LIKE '%女%' THEN 1.0 " +
                        "            WHEN keywords LIKE '%中性%' THEN 0.5 " +
                        "            ELSE 0.2 " +
                        "        END AS category_weight, " +
                        "        CASE " +
                        "            WHEN keywords LIKE '%男%' THEN '男' " +
                        "            WHEN keywords LIKE '%女%' THEN '女' " +
                        "            ELSE '未识别' " +
                        "        END AS gender " +
                        "    FROM dwd_ua " +
                        "    GROUP BY item_id, TUMBLE(behavior_time, INTERVAL '10' MINUTE), keywords " +
                        "), " +
                        "a3 AS ( " +
                        "    SELECT " +
                        "        user_id, " +
                        "        COUNT(behavior_type) AS act_count " +
                        "    FROM dwd_ua " +
                        "    GROUP BY user_id " +
                        "), " +
                        "a4 AS ( " +
                        "    SELECT " +
                        "        a1.user_id, " +
                        "        a2.gender, " +
                        "        a1.window_start, " +
                        "        ROUND(SUM(a1.time_decay * a1.behavior_weight * a2.category_weight * a3.act_count), 2) AS total_score, " +
                        "        MAX(a1.behavior_time) AS latest_time " +
                        "    FROM a1 " +
                        "    JOIN a2 ON a1.item_id = a2.item_id AND a1.window_start = a2.window_start " +
                        "    JOIN a3 ON a1.user_id = a3.user_id " +
                        "    GROUP BY a1.user_id, a2.gender, a1.window_start " +
                        "), " +
                        "max_scores AS ( " +
                        "    SELECT " +
                        "        user_id, " +
                        "        MAX(total_score) AS max_score " +
                        "    FROM a4 " +
                        "    GROUP BY user_id " +
                        "), " +
                        "latest_records AS ( " +
                        "    SELECT " +
                        "        user_id, " +
                        "        gender, " +
                        "        window_start, " +
                        "        latest_time " +
                        "    FROM ( " +
                        "        SELECT " +
                        "            user_id, " +
                        "            gender, " +
                        "            window_start, " +
                        "            latest_time, " +
                        "            ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY latest_time DESC) AS rn " +
                        "        FROM a4 " +
                        "    ) WHERE rn = 1 " +
                        ") " +
                        "SELECT " +
                        "    a4.user_id, " +
                        "    a4.window_start, " +
                        "    LAST_VALUE( " +
                        "        CASE WHEN a4.total_score * 2 < ms.max_score THEN a4.gender ELSE lr.gender END " +
                        "    ) AS gender " +
                        "FROM a4 " +
                        "JOIN max_scores ms ON a4.user_id = ms.user_id " +
                        "LEFT JOIN latest_records lr ON a4.user_id = lr.user_id " +
                        "where lr.gender is not null " +
                        "GROUP BY a4.user_id, a4.window_start"
        );

// 3. 输出结果（根据实际需求选择合适的方式）
        tEnv.toChangelogStream(result).print();

        env.execute();
    }
}
